package com.bff.gaia.mix.api.xjoin.client.hbase.all;

import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.collections.map.HashedMap;
import com.bff.gaia.api.java.typeutils.RowTypeInfo;
import com.bff.gaia.mix.api.xjoin.client.hbase.core.table.HbaseSideTableInfo;
import com.bff.gaia.mix.api.xjoin.core.enums.EJoinType;
import com.bff.gaia.mix.api.xjoin.core.side.AllReqRow;
import com.bff.gaia.mix.api.xjoin.core.side.FieldInfo;
import com.bff.gaia.mix.api.xjoin.core.side.JoinInfo;
import com.bff.gaia.mix.api.xjoin.core.side.SideTableInfo;
import com.bff.gaia.shaded.guava18.com.google.common.collect.Lists;
import com.bff.gaia.shaded.guava18.com.google.common.collect.Maps;
import com.bff.gaia.types.Row;
import com.bff.gaia.util.Collector;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.sql.SQLException;
import java.sql.Timestamp;
import java.util.Calendar;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicReference;

public class HbaseAllReqRow extends AllReqRow {

	private static final Logger LOG = LoggerFactory.getLogger(HbaseAllReqRow.class);

	private String tableName;

	private String tableParent;

	private boolean enableQualifierQuery;

	//	private AtomicReference<Map<String, Map<String, Object>>> cacheRef = new AtomicReference<>();
	private AtomicReference<Map<String, List<Map<String, Object>>>> cacheRef = new AtomicReference<>();

	public HbaseAllReqRow(RowTypeInfo rowTypeInfo, JoinInfo joinInfo, List<FieldInfo> outFieldInfoList, SideTableInfo sideTableInfo) {
		super(new HbaseAllSideInfo(rowTypeInfo, joinInfo, outFieldInfoList, sideTableInfo));
		tableName = ((HbaseSideTableInfo) sideTableInfo).getTableName();
		tableParent = ((HbaseSideTableInfo) sideTableInfo).getFamily();
		HbaseSideTableInfo hbaseSideTableInfo = (HbaseSideTableInfo) sideTableInfo;
		this.enableQualifierQuery = hbaseSideTableInfo.isQualifierQuery();
	}

	@Override
	public Row fillData(Row input, Object sideInput) {
		Map<String, Object> sideInputList = (Map<String, Object>) sideInput;
		Row row = new Row(sideInfo.getOutFieldInfoList().size());
		for (Map.Entry<Integer, Integer> entry : sideInfo.getInFieldIndex().entrySet()) {
			Object obj = input.getField(entry.getValue());
//			boolean isTimeIndicatorTypeInfo = TimeIndicatorTypeInfo.class.isAssignableFrom(xjoinInfo.getRowTypeInfo().getTypeAt(entry.getValue()).getClass());
			boolean isTimeIndicatorTypeInfo = false;
			//Type information for indicating event or processing time. However, it behaves like a regular SQL timestamp but is serialized as Long.
			if (obj instanceof Timestamp && isTimeIndicatorTypeInfo) {
				obj = ((Timestamp) obj).getTime();
			}
			row.setField(entry.getKey(), obj);
		}

		for (Map.Entry<Integer, Integer> entry : sideInfo.getSideFieldIndex().entrySet()) {
			if (sideInputList == null) {
				row.setField(entry.getKey(), null);
			} else {
				String key = sideInfo.getSideFieldNameIndex().get(entry.getKey());
				row.setField(entry.getKey(), sideInputList.get(tableParent + ":" + key));
			}
		}
//		System.out.println(row);
		return row;
	}

	@Override
	protected void initCache() throws SQLException {
//		Map<String, Map<String, Object>> newCache = Maps.newConcurrentMap();
		Map<String, List<Map<String, Object>>> newCache = Maps.newConcurrentMap();
		cacheRef.set(newCache);
		loadData(newCache);
		LOG.info("----- HBase all cacheRef init end:{}", Calendar.getInstance());
	}

	@Override
	protected void reloadCache() {
//		Map<String, Map<String, Object>> newCache = Maps.newConcurrentMap();
		Map<String, List<Map<String, Object>>> newCache = Maps.newConcurrentMap();
		try {
			loadData(newCache);
		} catch (SQLException e) {
			LOG.error("", e);
		}

		cacheRef.set(newCache);
		LOG.info("----- HBase all cacheRef reload end:{}", Calendar.getInstance());
	}

	@Override
	public void flatMap(Row value, Collector<Row> out) throws Exception {
		String rowKeyStr;
		if (enableQualifierQuery) {
			rowKeyStr = buildKey(value);
		} else {
			Map<String, Object> refData = Maps.newHashMap();
			for (int i = 0; i < sideInfo.getEqualValIndex().size(); i++) {
				Integer conValIndex = sideInfo.getEqualValIndex().get(i);
				Object equalObj = value.getField(conValIndex);
				if (equalObj == null) {
					out.collect(null);
				}
				refData.put(sideInfo.getEqualFieldList().get(i), equalObj);
			}
			rowKeyStr = ((HbaseAllSideInfo) sideInfo).getRowKeyBuilder().getRowKey(refData);
		}
//		System.out.println(rowKeyStr);

		List<Map<String, Object>> cacheList = cacheRef.get().get(rowKeyStr);
		for (Map<String, Object> one : cacheList) {
			out.collect(Row.project(fillData(value, one), sideInfo.getOutFieldIndexs()));
		}
	}

	private void loadData(Map<String, List<Map<String, Object>>> tmpCache) throws SQLException {
		SideTableInfo sideTableInfo = sideInfo.getSideTableInfo();
		HbaseSideTableInfo hbaseSideTableInfo = (HbaseSideTableInfo) sideTableInfo;
		Configuration conf = new Configuration();
		conf.set("hbase.zookeeper.quorum", hbaseSideTableInfo.getHost());
		Connection conn = null;
		Table table = null;
		ResultScanner resultScanner = null;
		try {
			conn = ConnectionFactory.createConnection(conf);
			table = conn.getTable(TableName.valueOf(tableName));
			resultScanner = table.getScanner(new Scan());
			for (Result r : resultScanner) {
				Map<String, Object> kv = new HashedMap();

				for (Cell cell : r.listCells()) {
					String family = Bytes.toString(CellUtil.cloneFamily(cell));
					String qualifier = Bytes.toString(CellUtil.cloneQualifier(cell));
					String value = Bytes.toString(CellUtil.cloneValue(cell));
					StringBuilder key = new StringBuilder();
					key.append(family).append(":").append(qualifier);
					kv.put(key.toString(), value);
				}
				String rowKeyStr;
				if (enableQualifierQuery) {
					rowKeyStr = buildKey(kv);
				} else {
					rowKeyStr = new String(r.getRow());
				}
				List<Map<String, Object>> list = tmpCache.computeIfAbsent(rowKeyStr, key -> Lists.newArrayList());
				list.add(kv);
			}
		} catch (IOException e) {
			e.printStackTrace();
		} finally {
			try {
				conn.close();
				table.close();
				resultScanner.close();
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
	}

	//QualifierQuery缓存主键构造
	private String buildKey(Row val) {
		StringBuilder sb = new StringBuilder("");
		for (int equalField : sideInfo.getEqualValIndex()) {
			sb.append(val.getField(equalField)).append("_");
		}

		return sb.toString();
	}

	private String buildKey(Map<String, Object> val) {
		StringBuilder sb = new StringBuilder("");
		for (String equalField : sideInfo.getEqualFieldList()) {
			sb.append(val.get(tableParent + ":" + equalField)).append("_");
		}

		return sb.toString();
	}
}